package com.wudl.realtime.avro;

import com.wudl.realtime.config.QuotConfig;
import org.apache.avro.io.BinaryDecoder;
import org.apache.avro.io.DecoderFactory;
import org.apache.avro.specific.SpecificDatumReader;
import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;

import java.io.ByteArrayInputStream;
import java.io.IOException;

/**
 * @author ：wudl
 * @date ：Created in 2022-01-06 0:17
 * @description：
 * @modified By： 1.创建泛型反序列化类实现反序列化接口
 * @version: 1.0
 */

public class AvroDeserializationSchema <T> implements DeserializationSchema<T> {

    private String topicName;

    //2.创建构造方法
    public AvroDeserializationSchema(String topicName) {
        this.topicName = topicName;
    }

    @Override
    public T deserialize(byte[] message) throws IOException {

        SpecificDatumReader<T> specificDatumReader = null;
        if (topicName.equals(QuotConfig.config.getProperty("sse.topic"))) {
            specificDatumReader = new SpecificDatumReader<>(new SseAvro().getSchema());
        } else {
            specificDatumReader = new SpecificDatumReader<>(new SzseAvro().getSchema());
        }
        ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(message);
        BinaryDecoder binaryDecoder = DecoderFactory.get().binaryDecoder(byteArrayInputStream, null);
        return specificDatumReader.read(null, binaryDecoder);
    }

    @Override
    public boolean isEndOfStream(T t) {
        return false;
    }

    @Override
    public TypeInformation<T> getProducedType() {
        if (topicName.equals(QuotConfig.config.getProperty("sse.topic"))) {
            return (TypeInformation<T>) TypeInformation.of(new TypeHint<SseAvro>() {
            });
        } else {
            return (TypeInformation<T>) TypeInformation.of(new TypeHint<SzseAvro>() {
            });
        }
    }
}
